package cn.tianyigps.obd.template;

import cn.tianyigps.obd.beans.StandardModel;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * Created with IDEA
 *
 * @description: kafka模板类
 * @author: cui.guibin
 * @create: 2019/4/3 16:41
 **/
public class KafkaTemplate {

    private static  Properties props = new Properties();
    private static  KafkaProducer<String, String> producer;
    static {
        props.put("bootstrap.servers", "172.19.25.119:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("max.poll.records", 1000);
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());

    }
    public static KafkaProducer<String, String> producer(){
        producer = new KafkaProducer<String, String>(props);
        return producer;
    }
    public static KafkaConsumer<String, String> consumer(){
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        return consumer;
    }

    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = consumer();
        consumer.subscribe(Arrays.asList("M7_DATA"));
        for (;;){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(1);
            if (null != consumerRecords && consumerRecords.count()>0){
                for (ConsumerRecord consumerRecord:consumerRecords) {
                    System.out.println(consumerRecord.value());
                }
            }

        }
    }
//    public static JavaType getCollectionType(Class<?> collectionClass, Class<?>... elementClasses) {
//        ObjectMapper objectMapper = new ObjectMapper();
//        return objectMapper.getTypeFactory().constructParametricType(collectionClass, elementClasses);
//    }

}
